package com.zimmor.mq.rocket.transaction;

import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author:zimmor
 * @Date: 2021/11/18 14:34
 * @Description 发送同步消息
 * @version: 1.0.0
 */

public class Producer {
    public static void main(String[] args) throws Exception {
       //发送消息的步骤
       //1.创建一个producer 设置producer的生产者组名groupID，
        TransactionMQProducer producer = new TransactionMQProducer("group6");
        //2.设置namesrv地址
        producer.setNamesrvAddr("192.168.204.128:9876;192.168.204.129:9876");
        //添加事务监听器
//        new AtomicInteger().getAndIncrement()
        producer.setTransactionListener(new TransactionListener() {
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                if("tag1".equals(message.getTags())) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                } else if("tag2".equals(message.getTags())) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                } else {
                    return LocalTransactionState.UNKNOW;
                }
            }

            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                if("tag3".equals(messageExt.getTags())) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                return LocalTransactionState.UNKNOW;
            }
        });
       //启动producer
        producer.start();

        for (int i = 0; i < 5; i++) {
            //4.新建消息对象。并设置topic和tag和消息体
            Message message = new Message("transactionTopic", "tag" + i, ("hello RocketMQ" + i).getBytes());
            //5.发送消息
            SendResult result = producer.sendMessageInTransaction(message, null);
            System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + "发送完毕");
            System.out.println(result.getSendStatus());
        }
       //6.关闭producer
//        producer.shutdown();
    }
}
